package com.fary.web.context.request.async;

import com.fary.core.task.AsyncTaskExecutor;
import com.fary.core.task.SimpleAsyncTaskExecutor;
import com.fary.core.task.SyncTaskExecutor;
import com.fary.util.Assert;
import com.fary.web.context.request.RequestAttributes;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.servlet.http.HttpServletRequest;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;

/**
 * 管理异步请求的核心类，主要是作为SPI,并不会被应用类直接使用。
 * 一个异步场景开始于一个线程（T1）中的普通request请求, 并发请求的处理可以通过调用startCallableProcessing来开启，或者调用startDeferredResultProcessing，以上两者都会在另外一个线程（T2）成产生结果。该结果将会被保存，并且请求会被派发到容器，然后在第三个（T3）线程中继续处理保存的结果。在这个派发线程中，刚才保存的结果可以通过getConcurrentResult来访问，或者通过hasConcurrentResult来判断结果是否存在。
 **/
public final class WebAsyncManager {

    // 没有结果的标识
    private static final Object RESULT_NONE = new Object();

    // 默认任务执行器
    private static final AsyncTaskExecutor DEFAULT_TASK_EXECUTOR = new SimpleAsyncTaskExecutor(WebAsyncManager.class.getSimpleName());

    private static final Log logger = LogFactory.getLog(WebAsyncManager.class);

    // Callable类型的超时拦截器
    private static final CallableProcessingInterceptor timeoutCallableInterceptor = new TimeoutCallableProcessingInterceptor();

    // DeferredResult类型结果的超时拦截器
    private static final DeferredResultProcessingInterceptor timeoutDeferredResultInterceptor =
            new TimeoutDeferredResultProcessingInterceptor();

    private static Boolean taskExecutorWarning = true;

    // 异步请求对象
    private AsyncWebRequest asyncWebRequest;

    // 执行器
    private AsyncTaskExecutor taskExecutor = DEFAULT_TASK_EXECUTOR;

    // 异步执行结果
    private volatile Object concurrentResult = RESULT_NONE;

    // 异步执行结果上下文
    private volatile Object[] concurrentResultContext;

    /*
     * Whether the concurrentResult is an error. If such errors remain unhandled, some
     * Servlet containers will call AsyncListener#onError at the end, after the ASYNC
     * and/or the ERROR dispatch (Boot's case), and we need to ignore those.
     */
    private volatile boolean errorHandlingInProgress;

    // 外部注册进来的Callable类型结果的拦截器列表
    private final Map<Object, CallableProcessingInterceptor> callableInterceptors = new LinkedHashMap<>();

    // 外部注册进来的DeferredResult类型结果的拦截器列表
    private final Map<Object, DeferredResultProcessingInterceptor> deferredResultInterceptors = new LinkedHashMap<>();

    WebAsyncManager() {
    }

    public void setAsyncWebRequest(AsyncWebRequest asyncWebRequest) {
        Assert.notNull(asyncWebRequest, "AsyncWebRequest must not be null");
        this.asyncWebRequest = asyncWebRequest;
        this.asyncWebRequest.addCompletionHandler(
                () -> asyncWebRequest.removeAttribute(WebAsyncUtils.WEB_ASYNC_MANAGER_ATTRIBUTE, RequestAttributes.SCOPE_REQUEST));
    }

    /**
     * 设置异步请求处理器
     */
    public void setTaskExecutor(AsyncTaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    public boolean isConcurrentHandlingStarted() {
        return (this.asyncWebRequest != null && this.asyncWebRequest.isAsyncStarted());
    }

    /**
     * 是否有返回结果
     */
    public boolean hasConcurrentResult() {
        return (this.concurrentResult != RESULT_NONE);
    }

    /**
     * 获取异步执行返回结果
     */
    public Object getConcurrentResult() {
        return this.concurrentResult;
    }

    /**
     * 获取异步执行上下文
     */
    public Object[] getConcurrentResultContext() {
        return this.concurrentResultContext;
    }

    /**
     * 获取拦截器
     */
    public CallableProcessingInterceptor getCallableInterceptor(Object key) {
        return this.callableInterceptors.get(key);
    }

    public DeferredResultProcessingInterceptor getDeferredResultInterceptor(Object key) {
        return this.deferredResultInterceptors.get(key);
    }

    public void registerCallableInterceptor(Object key, CallableProcessingInterceptor interceptor) {
        Assert.notNull(key, "Key is required");
        Assert.notNull(interceptor, "CallableProcessingInterceptor  is required");
        this.callableInterceptors.put(key, interceptor);
    }

    public void registerCallableInterceptors(CallableProcessingInterceptor... interceptors) {
        Assert.notNull(interceptors, "A CallableProcessingInterceptor is required");
        for (CallableProcessingInterceptor interceptor : interceptors) {
            String key = interceptor.getClass().getName() + ":" + interceptor.hashCode();
            this.callableInterceptors.put(key, interceptor);
        }
    }

    public void registerDeferredResultInterceptor(Object key, DeferredResultProcessingInterceptor interceptor) {
        Assert.notNull(key, "Key is required");
        Assert.notNull(interceptor, "DeferredResultProcessingInterceptor is required");
        this.deferredResultInterceptors.put(key, interceptor);
    }

    public void registerDeferredResultInterceptors(DeferredResultProcessingInterceptor... interceptors) {
        Assert.notNull(interceptors, "A DeferredResultProcessingInterceptor is required");
        for (DeferredResultProcessingInterceptor interceptor : interceptors) {
            String key = interceptor.getClass().getName() + ":" + interceptor.hashCode();
            this.deferredResultInterceptors.put(key, interceptor);
        }
    }

    public void clearConcurrentResult() {
        synchronized (WebAsyncManager.this) {
            this.concurrentResult = RESULT_NONE;
            this.concurrentResultContext = null;
        }
    }

    /**
     * 处理并使用  AsyncTaskExecutor 执行给定任务。
     * 保存任务执行的结果并分派请求以恢复对该结果的处理。如果任务引发异常，则保存的结果将是引发的异常。
     */
    @SuppressWarnings({"rawtypes", "unchecked"})
    public void startCallableProcessing(Callable<?> callable, Object... processingContext) throws Exception {
        Assert.notNull(callable, "Callable must not be null");
        startCallableProcessing(new WebAsyncTask(callable), processingContext);
    }

    public void startCallableProcessing(final WebAsyncTask<?> webAsyncTask, Object... processingContext) throws Exception {

        Assert.notNull(webAsyncTask, "WebAsyncTask must not be null");
        Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null");

        // 超时时间
        Long timeout = webAsyncTask.getTimeout();
        if (timeout != null) {
            this.asyncWebRequest.setTimeout(timeout);
        }

        // 异步线程池
        AsyncTaskExecutor executor = webAsyncTask.getExecutor();
        if (executor != null) {
            this.taskExecutor = executor;
        } else {
            logExecutorWarning();
        }

        List<CallableProcessingInterceptor> interceptors = new ArrayList<>();
        interceptors.add(webAsyncTask.getInterceptor());
        interceptors.addAll(this.callableInterceptors.values());
        interceptors.add(timeoutCallableInterceptor);

        // 异步任务
        final Callable<?> callable = webAsyncTask.getCallable();
        final CallableInterceptorChain interceptorChain = new CallableInterceptorChain(interceptors);

        // 异步超时回调处理
        this.asyncWebRequest.addTimeoutHandler(() -> {
            if (logger.isDebugEnabled()) {
                logger.debug("Async request timeout for " + formatRequestUri());
            }
            Object result = interceptorChain.triggerAfterTimeout(this.asyncWebRequest, callable);
            if (result != CallableProcessingInterceptor.RESULT_NONE) {
                setConcurrentResultAndDispatch(result);
            }
        });

        // 异步异常回调处理
        this.asyncWebRequest.addErrorHandler(ex -> {
            if (!this.errorHandlingInProgress) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Async request error for " + formatRequestUri() + ": " + ex);
                }
                Object result = interceptorChain.triggerAfterError(this.asyncWebRequest, callable, ex);
                result = (result != CallableProcessingInterceptor.RESULT_NONE ? result : ex);
                setConcurrentResultAndDispatch(result);
            }
        });

        // 异步完成回调处理
        this.asyncWebRequest.addCompletionHandler(() ->
                interceptorChain.triggerAfterCompletion(this.asyncWebRequest, callable));

        //开启异步前置处理
        interceptorChain.applyBeforeConcurrentHandling(this.asyncWebRequest, callable);
        // 开始异步处理
        // 调用asyncWebRequest.startAsync()开启异步请求
        startAsyncProcessing(processingContext);
        try {
            Future<?> future = this.taskExecutor.submit(() -> {
                Object result = null;
                try {
                    // 在异步线程中，callable执行之前，执行applyPreProcess
                    interceptorChain.applyPreProcess(this.asyncWebRequest, callable);
                    result = callable.call();
                } catch (Throwable ex) {
                    result = ex;
                } finally {
                    // 异步执行后置处理
                    result = interceptorChain.applyPostProcess(this.asyncWebRequest, callable, result);
                }
                // 设置当前的结果并转发
                setConcurrentResultAndDispatch(result);
            });
            // 往拦截器链中设置Future
            interceptorChain.setTaskFuture(future);
        } catch (RejectedExecutionException ex) {
            Object result = interceptorChain.applyPostProcess(this.asyncWebRequest, callable, ex);
            setConcurrentResultAndDispatch(result);
            throw ex;
        }
    }

    private void logExecutorWarning() {
        if (taskExecutorWarning && logger.isWarnEnabled()) {
            synchronized (DEFAULT_TASK_EXECUTOR) {
                AsyncTaskExecutor executor = this.taskExecutor;
                if (taskExecutorWarning &&
                        (executor instanceof SimpleAsyncTaskExecutor || executor instanceof SyncTaskExecutor)) {
                    String executorTypeName = executor.getClass().getSimpleName();
                    logger.warn("\n!!!\n" +
                            "An Executor is required to handle java.util.concurrent.Callable return values.\n" +
                            "Please, configure a TaskExecutor in the MVC config under \"async support\".\n" +
                            "The " + executorTypeName + " currently in use is not suitable under load.\n" +
                            "-------------------------------\n" +
                            "Request URI: '" + formatRequestUri() + "'\n" +
                            "!!!");
                    taskExecutorWarning = false;
                }
            }
        }
    }

    private String formatRequestUri() {
        HttpServletRequest request = this.asyncWebRequest.getNativeRequest(HttpServletRequest.class);
        return request != null ? request.getRequestURI() : "servlet container";
    }

    private void setConcurrentResultAndDispatch(Object result) {
        synchronized (WebAsyncManager.this) {
            if (this.concurrentResult != RESULT_NONE) {
                return;
            }
            this.concurrentResult = result;
            this.errorHandlingInProgress = (result instanceof Throwable);
        }

        if (this.asyncWebRequest.isAsyncComplete()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Async result set but request already complete: " + formatRequestUri());
            }
            return;
        }

        if (logger.isDebugEnabled()) {
            boolean isError = result instanceof Throwable;
            logger.debug("Async " + (isError ? "error" : "result set") + ", dispatch to " + formatRequestUri());
        }
        this.asyncWebRequest.dispatch();
    }

    /**
     *
     */
    public void startDeferredResultProcessing(
            final DeferredResult<?> deferredResult, Object... processingContext) throws Exception {

        Assert.notNull(deferredResult, "DeferredResult must not be null");
        Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null");

        Long timeout = deferredResult.getTimeoutValue();
        if (timeout != null) {
            this.asyncWebRequest.setTimeout(timeout);
        }

        List<DeferredResultProcessingInterceptor> interceptors = new ArrayList<>();
        interceptors.add(deferredResult.getInterceptor());
        interceptors.addAll(this.deferredResultInterceptors.values());
        interceptors.add(timeoutDeferredResultInterceptor);

        final DeferredResultInterceptorChain interceptorChain = new DeferredResultInterceptorChain(interceptors);

        this.asyncWebRequest.addTimeoutHandler(() -> {
            try {
                interceptorChain.triggerAfterTimeout(this.asyncWebRequest, deferredResult);
            } catch (Throwable ex) {
                setConcurrentResultAndDispatch(ex);
            }
        });

        this.asyncWebRequest.addErrorHandler(ex -> {
            if (!this.errorHandlingInProgress) {
                try {
                    if (!interceptorChain.triggerAfterError(this.asyncWebRequest, deferredResult, ex)) {
                        return;
                    }
                    deferredResult.setErrorResult(ex);
                } catch (Throwable interceptorEx) {
                    setConcurrentResultAndDispatch(interceptorEx);
                }
            }
        });

        this.asyncWebRequest.addCompletionHandler(()
                -> interceptorChain.triggerAfterCompletion(this.asyncWebRequest, deferredResult));

        interceptorChain.applyBeforeConcurrentHandling(this.asyncWebRequest, deferredResult);
        startAsyncProcessing(processingContext);

        try {
            interceptorChain.applyPreProcess(this.asyncWebRequest, deferredResult);
            deferredResult.setResultHandler(result -> {
                result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
                setConcurrentResultAndDispatch(result);
            });
        } catch (Throwable ex) {
            setConcurrentResultAndDispatch(ex);
        }
    }

    private void startAsyncProcessing(Object[] processingContext) {
        synchronized (WebAsyncManager.this) {
            this.concurrentResult = RESULT_NONE;
            this.concurrentResultContext = processingContext;
            this.errorHandlingInProgress = false;
        }
        this.asyncWebRequest.startAsync();

        if (logger.isDebugEnabled()) {
            logger.debug("Started async request");
        }
    }

}